
package cn.junechiu.learnrxjava;

import android.os.Bundle;
import android.os.SystemClock;
import android.support.annotation.Nullable;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * Created by android on 2017/7/19.
 * <p>
 * 1.组合的过程是分别从 两根水管里各取出一个事件 来进行组合, 并且一个事件只能被使用一次, 组合的顺序是严格按照事件发送的顺利 来进行的,
 * 也就是说不会出现圆形1 事件和三角形B 事件进行合并, 也不可能出现圆形2 和三角形A 进行合并的情况.
 * <p>
 * 2.最终下游收到的事件数量 是和上游中发送事件最少的那一根水管的事件数量 相同. 这个也很好理解, 因为是从每一根水管 里取一个事件来进行合并, 最少的
 * 那个肯定就最先取完 , 这个时候其他的水管尽管还有事件 , 但是已经没有足够的事件来组合了, 因此下游就不会收到剩余的事件了. 作者：Season_zlc
 * 链接：http://www.jianshu.com/p/bb58571cdb64
 * <p>
 * 比如一个界面需要展示用户的一些信息, 而这些信息分别要从两个服务器接口中获取, 而只有当两个都获取到了之后才能进行展示, 这个时候就可以用Zip了:
 */
public class ZipActivity extends AppCompatActivity {

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
//        sample1();
//        sample2();
        sample3();
    }

    public void sample1() {
        Observable observable1 = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter emitter) throws Exception {
                emitter.onNext("A");
                Log.d("observable1", "emitterA");
                emitter.onNext("B");
                Log.d("observable1", "emitterB");
                emitter.onNext("C");
                Log.d("observable1", "emitterC");
                emitter.onNext("D");
                Log.d("observable1", "emitterD");
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io()); // io线程中

        Observable observable2 = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter emitter) throws Exception {
                emitter.onNext(1);
                Log.d("observable2", "emitter1");
                emitter.onNext(1);
                Log.d("observable2", "emitter2");
                emitter.onNext(1);
                Log.d("observable2", "emitter3");
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io());// io线程中

        Observable.zip(observable1, observable2, new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s1, Integer integer) throws Exception {
                return integer + s1;
            }
        }).subscribe(new Observer() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Object value) {
                Log.d("ZipActivity", "onNext: " + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d("ZipActivity", "onError: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.d("ZipActivity", "onComplete: ");
            }
        });
    }

    //同步,同一个线程中通信，下游处理完事件后，上游才能发送,没有水缸存储信息
    public void sample2() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; ; i++) {
                    e.onNext(i);
                }
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                SystemClock.sleep(2000);
                Log.d("ZipActivity", "sample2: " + integer);
            }
        });
    }

    //异步，不在同一个线程中,有水缸存储数据，内存增大
    public void sample3() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; ; i++) {
                    e.onNext(i);
                }
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer s) throws Exception {
                SystemClock.sleep(2000);
                Log.d("ZipActivity", "sample3: " + s);
            }
        });
    }
}
